package cn.itcast;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.List;

public class Demo {

    // 创建表,一般不用java代码（不方便）
    @Test
    public void test01() throws Exception{

        // 1) 创建HBase连接对象:

        //Configuration conf = new Configuration();
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum","node1:2181");
        Connection hbConn = ConnectionFactory.createConnection(conf);

        // 2) 从连接对象中获取相关的管理对象: Admin(对表的操作) 和 Table(对表数据操作)
        Admin admin = hbConn.getAdmin();

        // 3) 执行相关的操作

        boolean flag = admin.tableExists(TableName.valueOf("test6")); // 为true 表示存在, 否则为不存在

        if(!flag){
            // 说明 表不存在
            //3.1: 创建 表的构建器对象
            TableDescriptorBuilder descBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf("test6"));


            //3.2: 在构建器 设置表的列族信息
            ColumnFamilyDescriptorBuilder familyDesc = ColumnFamilyDescriptorBuilder.newBuilder("c1".getBytes());
            descBuilder.setColumnFamily(familyDesc.build());

            //3.3: 基于表构建器  构建表基本信息封装对象
            TableDescriptor desc = descBuilder.build();

            admin.createTable(desc);
        }

        // 4) 释放资源
        admin.close();
        hbConn.close();

    }

    // 添加数据
    @Test
    public void test02() throws Exception{

        //1) 创建Hbase的连接对象
        Configuration conf = HBaseConfiguration.create();

        conf.set("hbase.zookeeper.quorum","node1:2181");

        Connection hbConn = ConnectionFactory.createConnection(conf);
        //2) 通过连接对象 获取相关的管理对象: admin  和 table

        Table table = hbConn.getTable(TableName.valueOf("test6"));
        //3) 执行相关的操作:  添加数据

        Put put = new Put(Bytes.toBytes("2"));

        put.addColumn("c1".getBytes(),"NAME".getBytes(),"唐舞麟".getBytes());
        put.addColumn("c1".getBytes(),"ADDRESS".getBytes(),"贵州省铜仁市德江县7单元267室".getBytes());
        put.addColumn("c1".getBytes(),"SEX".getBytes(),"男".getBytes());
        put.addColumn("c1".getBytes(),"PAY_TIME".getBytes(),"2020-05-10".getBytes());


        table.put(put);

        //4) 处理结果集:  -- 此步骤不需要

        //5) 释放资源
        table.close();
        hbConn.close();


    }

    private Connection hbConn;
    private Table table;
    private Admin admin;
    private String tableName = "test6";
    @Before
    public void before() throws Exception{

        //1) 创建Hbase的连接对象
        Configuration conf = HBaseConfiguration.create();

        conf.set("hbase.zookeeper.quorum","node1:2181");

        hbConn = ConnectionFactory.createConnection(conf);
        //2) 通过连接对象 获取相关的管理对象: admin  和 table
        admin = hbConn.getAdmin();
        table = hbConn.getTable(TableName.valueOf(tableName));


    }

    // 根据rowkey 查询某一条数据
    @Test
    public void test03() throws Exception{

        // 3) 执行相关的操作 get

        Get get = new Get(Bytes.toBytes("2"));
        Result result = table.get(get); // 一个Result 表示 就是一行数据

        //4) 处理结果集

        //4.1: 获取一行数据中每一个单元格
        List<Cell> listCells = result.listCells();

        //4.1.1获取rowkey
        String rowkey = Bytes.toString(result.getRow());
        System.out.println("rowkey为:"+ rowkey);

        //4.2: 遍历单元格, 从单元格获取数据
        for (Cell cell : listCells) {

            // 从单元格可以获取那些内容: rowkey  列族 列名 列值

            // 4.2.2 获取 列族
            byte[] familyBytes = CellUtil.cloneFamily(cell);
            String family = Bytes.toString(familyBytes);
            // 4.2.3 获取 列名(列限定符)
            byte[] qualifierBytes = CellUtil.cloneQualifier(cell);
            String qualifier = Bytes.toString(qualifierBytes);
            // 4.2.4 获取 列值
            byte[] valueBytes = CellUtil.cloneValue(cell);
            String value = Bytes.toString(valueBytes);


            System.out.println("列族为:"+family+"; 列名为:"+qualifier+"; 列值为:"+value);

        }


    }

    // 删除数据操作:
    @Test
    public  void  test04() throws Exception{

        // 3) 执行相关的操作 : delete  和 deleteall
        Delete delete = new Delete(Bytes.toBytes("1"));

//        delete.addColumn("c1".getBytes(),"NAME".getBytes());
        delete.addFamily("c1".getBytes());
        table.delete(delete);


        //4) 处理结果集 -- 此步骤不需要

    }

    // 删除表操作:
    @Test
    public  void  test05() throws Exception{
        // 3) 执行相关的操作

        boolean flag = admin.isTableEnabled(TableName.valueOf(tableName)); // 判断是否启用表
        if(flag){

            admin.disableTable(TableName.valueOf(tableName)); //如果启用, 将其先禁用
        }

        admin.deleteTable(TableName.valueOf(tableName));

        //4) 处理结果集 -- 此步骤不需要
    }


    // 基于scan扫描:  需求: 查询 id为101-102的数据
    // select  *  from  test_table  where  id between 101  and 102 ;

    @Test
    public void test06() throws Exception{

        //3. 执行相关的操作
        Scan scan = new Scan();
        //scan查询是左闭右开，查询的结果数据，不包含最后设置的rowkey
        scan.withStartRow(Bytes.toBytes("1"));
        scan.withStopRow(Bytes.toBytes("3"));

        ResultScanner results = table.getScanner(scan);

        //4. 处理结果集
        //4.1: 获取每一行数据
        for (Result result : results) {

            //4.2 获取每一行的所有单元格数据
            List<Cell> listCells = result.listCells();
            String rowkey = Bytes.toString(result.getRow());
            System.out.println("rowkey为:"+ rowkey);

            //4.3 遍历每一个单元格, 获取数据
            for (Cell cell : listCells) {
                String qualifier  = Bytes.toString(CellUtil.cloneQualifier(cell));

                if("NAME".equals(qualifier)){

                    String vlaue = Bytes.toString(CellUtil.cloneValue(cell));
                    System.out.println("姓名为:"+ vlaue);
                }

                if("ADDRESS".equals(qualifier)){

                    String vlaue = Bytes.toString(CellUtil.cloneValue(cell));
                    System.out.println("地址为:"+ vlaue);
                }

                if("SEX".equals(qualifier)){

                    String vlaue = Bytes.toString(CellUtil.cloneValue(cell));
                    System.out.println("性别为:"+ vlaue);
                }

            }
            System.out.println("-----------------------------");


        }

    }


    @After
    public  void after() throws Exception{
        //5. 释放资源
        admin.close();
        table.close();
        hbConn.close();

    }


}